package com.nami.dispatcher.controller.middleware.mq.consumer;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.TimeInterval;
import com.google.protobuf.InvalidProtocolBufferException;
import com.nami.dispatcher.facade.pmsg.ForwardPrivateMsgFacade;
import com.nami.dispatcher.facade.pmsg.request.PChatRequestCmd;
import com.nami.gate.external.basic.bo.ConnectedUserBO;
import com.nami.gate.external.basic.constants.DispatcherTopicConst;
import com.nami.im.network.model.message.v2.Message;
import com.nami.transport.proto.codec.MqTransMsg;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.dromara.common.json.utils.JsonUtils;
import org.springframework.stereotype.Component;


@Slf4j
@Component
@AllArgsConstructor
@RocketMQMessageListener(topic = DispatcherTopicConst.PCHAT_TOPIC,
    consumerGroup = DispatcherTopicConst.PCHAT_CONSUMER_GROUP,
    consumeMode = ConsumeMode.CONCURRENTLY)
public class PrivateMsgConsumer implements RocketMQListener<MessageExt> {

    private final ForwardPrivateMsgFacade sendPrivateMsgFacade;

    @Override
    public void onMessage(MessageExt msg) {
        try {
            TimeInterval timer = DateUtil.timer();
//            log.info("[c2c] begin. message:{}", msg);
            PChatRequestCmd cmd = parsePrivateChatMessage(msg.getBody());
            sendPrivateMsgFacade.execute(cmd);
            Message message = cmd.getMessage();
            if (timer.interval() > 200) {
                log.warn("单聊消息出现处理慢，msgId={}，clientMsgId={}， use:{}ms",
                    message.getMsgId(), message.getClientUid(), timer.interval());
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    private PChatRequestCmd parsePrivateChatMessage(byte[] msg) throws InvalidProtocolBufferException {
        MqTransMsg.Logic2DispatchBody request = MqTransMsg.Logic2DispatchBody.parseFrom(msg);
        Message message = JsonUtils.parseObject(request.getMessage().toByteArray(), Message.class);
        PChatRequestCmd cmd = new PChatRequestCmd();
        cmd.setSuccess(request.getSuccess());
        cmd.setErrMessage(request.getErrMessage());
        cmd.setErrCode(request.getErrCode());
        cmd.setMessage(message);
        cmd.setQos(request.getQos());

        MqTransMsg.TransUser sender = request.getSender();
        ConnectedUserBO senderUserCO = new ConnectedUserBO();
        senderUserCO.setAppKey(sender.getAppKey());
        senderUserCO.setUserId(sender.getUserId());
        senderUserCO.setPlatform(sender.getPlatform());
        senderUserCO.setConnectId(sender.getConnectId());
        senderUserCO.setInstanceId(sender.getInstanceId());
        cmd.setSender(senderUserCO);

        return cmd;
    }

}
